/*
 *
 *  *  Copyright 2010-2016 OrientDB LTD (http://orientdb.com)
 *  *
 *  *  Licensed under the Apache License, Version 2.0 (the "License");
 *  *  you may not use this file except in compliance with the License.
 *  *  You may obtain a copy of the License at
 *  *
 *  *       http://www.apache.org/licenses/LICENSE-2.0
 *  *
 *  *  Unless required by applicable law or agreed to in writing, software
 *  *  distributed under the License is distributed on an "AS IS" BASIS,
 *  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  *  See the License for the specific language governing permissions and
 *  *  limitations under the License.
 *  *
 *  * For more information: http://orientdb.com
 *
 */

package com.orientechnologies.orient.client.remote.db.document;

import static com.orientechnologies.orient.core.storage.OStorage.LOCKING_STRATEGY.EXCLUSIVE_LOCK;

import com.orientechnologies.common.concur.lock.OLockException;
import com.orientechnologies.common.exception.OException;
import com.orientechnologies.common.log.OLogManager;
import com.orientechnologies.common.log.OLogger;
import com.orientechnologies.orient.client.remote.OLiveQueryClientListener;
import com.orientechnologies.orient.client.remote.ORemoteClient;
import com.orientechnologies.orient.client.remote.ORemoteClientSession;
import com.orientechnologies.orient.client.remote.ORemoteQueryResult;
import com.orientechnologies.orient.client.remote.message.OLockRecordResponse;
import com.orientechnologies.orient.client.remote.message.ORemoteResultSet;
import com.orientechnologies.orient.core.Orient;
import com.orientechnologies.orient.core.cache.OLocalRecordCache;
import com.orientechnologies.orient.core.command.OCommandOutputListener;
import com.orientechnologies.orient.core.command.script.OCommandScriptException;
import com.orientechnologies.orient.core.conflict.ORecordConflictStrategy;
import com.orientechnologies.orient.core.db.ODatabase;
import com.orientechnologies.orient.core.db.ODatabaseDocumentInternal;
import com.orientechnologies.orient.core.db.ODatabaseListener;
import com.orientechnologies.orient.core.db.ODatabaseRecordThreadLocal;
import com.orientechnologies.orient.core.db.OHookReplacedRecordThreadLocal;
import com.orientechnologies.orient.core.db.OLiveQueryMonitor;
import com.orientechnologies.orient.core.db.OLiveQueryResultListener;
import com.orientechnologies.orient.core.db.OrientDBConfig;
import com.orientechnologies.orient.core.db.document.ODatabaseDocument;
import com.orientechnologies.orient.core.db.document.ODatabaseDocumentAbstract;
import com.orientechnologies.orient.core.db.document.RecordReader;
import com.orientechnologies.orient.core.db.record.OIdentifiable;
import com.orientechnologies.orient.core.db.record.ORecordElement;
import com.orientechnologies.orient.core.exception.OCommandExecutionException;
import com.orientechnologies.orient.core.exception.ODatabaseException;
import com.orientechnologies.orient.core.exception.ORecordNotFoundException;
import com.orientechnologies.orient.core.fetch.OFetchHelper;
import com.orientechnologies.orient.core.hook.ORecordHook;
import com.orientechnologies.orient.core.id.ORID;
import com.orientechnologies.orient.core.id.ORecordId;
import com.orientechnologies.orient.core.index.OClassIndexManager;
import com.orientechnologies.orient.core.iterator.ORecordIteratorCluster;
import com.orientechnologies.orient.core.metadata.OSessionMetadata;
import com.orientechnologies.orient.core.metadata.schema.OClass;
import com.orientechnologies.orient.core.metadata.schema.OImmutableClass;
import com.orientechnologies.orient.core.metadata.schema.OSessionSchema;
import com.orientechnologies.orient.core.metadata.security.OImmutableUser;
import com.orientechnologies.orient.core.metadata.security.ORole;
import com.orientechnologies.orient.core.metadata.security.ORule;
import com.orientechnologies.orient.core.metadata.security.OUser;
import com.orientechnologies.orient.core.metadata.sequence.OSequenceAction;
import com.orientechnologies.orient.core.record.OEdge;
import com.orientechnologies.orient.core.record.ORecord;
import com.orientechnologies.orient.core.record.ORecordInternal;
import com.orientechnologies.orient.core.record.OVertex;
import com.orientechnologies.orient.core.record.impl.ODocument;
import com.orientechnologies.orient.core.record.impl.ODocumentInternal;
import com.orientechnologies.orient.core.record.impl.OEdgeDocument;
import com.orientechnologies.orient.core.record.impl.OVertexDocument;
import com.orientechnologies.orient.core.serialization.serializer.record.ORecordSerializerFactory;
import com.orientechnologies.orient.core.serialization.serializer.record.binary.ORecordSerializerNetworkV37Client;
import com.orientechnologies.orient.core.sql.executor.OResult;
import com.orientechnologies.orient.core.sql.executor.OResultSet;
import com.orientechnologies.orient.core.storage.OPhysicalPosition;
import com.orientechnologies.orient.core.storage.ORawBuffer;
import com.orientechnologies.orient.core.storage.ORecordMetadata;
import com.orientechnologies.orient.core.storage.OStorage;
import com.orientechnologies.orient.core.storage.OStorageInfo;
import com.orientechnologies.orient.core.storage.cluster.OOfflineClusterException;
import com.orientechnologies.orient.core.storage.ridbag.sbtree.OBonsaiCollectionPointer;
import com.orientechnologies.orient.core.storage.ridbag.sbtree.OSBTreeCollectionManager;
import com.orientechnologies.orient.core.tx.OTransaction;
import com.orientechnologies.orient.core.tx.OTransactionIndexChanges;
import com.orientechnologies.orient.core.tx.OTransactionInternal;
import com.orientechnologies.orient.core.tx.OTransactionOptimistic;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/** Created by tglman on 30/06/16. */
public class ODatabaseDocumentRemote extends ODatabaseDocumentAbstract {
  private static final OLogger logger =
      OLogManager.instance().logger(ODatabaseDocumentRemote.class);

  protected ORemoteClientSession session;
  private OrientDBConfig config;
  private ORemoteClient client;

  public ODatabaseDocumentRemote(OSharedContextRemote sharedContext) {
    activateOnCurrentThread();

    try {
      status = STATUS.CLOSED;
      this.client = sharedContext.getClient();
      // OVERWRITE THE URL
      url = this.client.getURL();
      this.sharedContext = sharedContext;
      this.componentsFactory = this.client.getComponentsFactory();

      unmodifiableHooks = Collections.unmodifiableMap(hooks);

      localCache = new OLocalRecordCache();
      this.sharedContext.startSession();
      init();

      databaseOwner = this;

    } catch (Exception t) {
      ODatabaseRecordThreadLocal.instance().remove();

      throw OException.wrapException(new ODatabaseException("Error on opening database "), t);
    }
  }

  @Override
  public <DB extends ODatabase> DB set(ATTRIBUTES iAttribute, Object iValue) {

    if (iAttribute == ATTRIBUTES.CUSTOM) {
      String stringValue = iValue.toString();
      int indx = stringValue != null ? stringValue.indexOf('=') : -1;
      if (indx < 0) {
        if ("clear".equalsIgnoreCase(stringValue)) {
          String query = "alter database CUSTOM 'clear'";
          // Bypass the database command for avoid transaction management
          ORemoteQueryResult result = getRemoteClient().command(this, query, new Object[] {iValue});
          result.getResult().close();
        } else
          throw new IllegalArgumentException(
              "Syntax error: expected <name> = <value> or clear, instead found: " + iValue);
      } else {
        String customName = stringValue.substring(0, indx).trim();
        String customValue = stringValue.substring(indx + 1).trim();
        setCustom(customName, customValue);
      }
    } else {
      String query = "alter database " + iAttribute.name() + " ? ";
      // Bypass the database command for avoid transaction management
      ORemoteQueryResult result = getRemoteClient().command(this, query, new Object[] {iValue});
      result.getResult().close();
      getRemoteClient().reload(getSession());
    }

    return (DB) this;
  }

  @Override
  public <DB extends ODatabase> DB setCustom(String name, Object iValue) {
    if ("clear".equals(name) && iValue == null) {
      String query = "alter database CUSTOM 'clear'";
      // Bypass the database command for avoid transaction management
      ORemoteQueryResult result = getRemoteClient().command(this, query, new Object[] {});
      result.getResult().close();
    } else {
      String query = "alter database CUSTOM  " + name + " = ?";
      // Bypass the database command for avoid transaction management
      ORemoteQueryResult result = getRemoteClient().command(this, query, new Object[] {iValue});
      result.getResult().close();
      getRemoteClient().reload(getSession());
    }
    return (DB) this;
  }

  public ODatabaseDocumentInternal copy() {
    ODatabaseDocumentRemote database =
        new ODatabaseDocumentRemote((OSharedContextRemote) this.sharedContext);
    database.client = client.copy(this, database);
    database.client.addUser();
    database.status = STATUS.OPEN;
    database.applyAttributes(config);
    database.initAtFirstOpen();
    database.user = this.user;
    this.activateOnCurrentThread();
    return database;
  }

  public void internalOpen(String user, String password, OrientDBConfig config) {
    this.config = config;
    applyAttributes(config);
    applyListeners(config);
    try {

      client.open(this.getSession(), user, password, config.getConfigurations());

      status = STATUS.OPEN;

      initAtFirstOpen();
      this.user =
          new OImmutableUser(
              -1, new OUser(user, password)); // .addRole(new ORole("passthrough", null,
      // ORole.ALLOW_MODES.ALLOW_ALL_BUT)));

      // WAKE UP LISTENERS
      callOnOpenListeners();

    } catch (OException e) {
      close();
      ODatabaseRecordThreadLocal.instance().remove();
      throw e;
    } catch (Exception e) {
      close();
      ODatabaseRecordThreadLocal.instance().remove();
      throw OException.wrapException(
          new ODatabaseException("Cannot open database url=" + getURL()), e);
    }
  }

  public void initPush(OMetadataPushListener handler) {
    getRemoteClient().initPush(getSession(), handler);
  }

  private void applyAttributes(OrientDBConfig config) {
    for (Entry<ATTRIBUTES, Object> attrs : config.getAttributes().entrySet()) {
      this.set(attrs.getKey(), attrs.getValue());
    }
  }

  private void initAtFirstOpen() {
    if (initialized) return;

    ORecordSerializerFactory serializerFactory = ORecordSerializerFactory.instance();
    serializer = serializerFactory.getFormat(ORecordSerializerNetworkV37Client.NAME);
    localCache.startup(this);
    componentsFactory = getRemoteClient().getComponentsFactory();
    user = null;

    loadMetadata();

    initialized = true;
  }

  @Override
  protected void loadMetadata() {
    metadata = new OSessionMetadata(this);
    metadata.init(sharedContext);
    sharedContext.load(this);
  }

  private void applyListeners(OrientDBConfig config) {
    for (ODatabaseListener listener : config.getListeners()) {
      registerListener(listener);
    }
  }

  public ODatabaseDocumentAbstract begin(final OTransaction.TXTYPE iType) {
    checkOpenness();
    checkIfActive();
    if (currentTx.isActive()) {
      if (iType == OTransaction.TXTYPE.OPTIMISTIC && currentTx instanceof OTransactionOptimistic) {
        currentTx.begin();
        return this;
      }
      currentTx.rollback(true, 0);
    }

    // CHECK IT'S NOT INSIDE A HOOK
    if (!inHook.isEmpty())
      throw new IllegalStateException("Cannot begin a transaction while a hook is executing");

    // WAKE UP LISTENERS
    for (ODatabaseListener listener : browseListeners())
      try {
        listener.onBeforeTxBegin(this);
      } catch (Exception t) {
        logger.error("Error before tx begin", t);
      } catch (Error e) {
        logger.error("Error before tx begin", e);
        throw e;
      }

    switch (iType) {
      case NOTX:
        setDefaultTransactionMode(null);
        break;
      case OPTIMISTIC:
        currentTx = new OTransactionOptimisticClient(this);
        break;
      case PESSIMISTIC:
        throw new UnsupportedOperationException("Pessimistic transaction");
    }
    currentTx.begin();
    return this;
  }

  public ORemoteClientSession getSession() {
    if (session == null) {
      session = client.newInitialSession();
    }
    return session;
  }

  public void setSessionMetadata(ORemoteClientSession sessionMetadata) {
    this.session = sessionMetadata;
  }

  @Override
  public OStorage getStorage() {
    throw new UnsupportedOperationException();
  }

  public ORemoteClient getRemoteClient() {
    return client;
  }

  @Override
  public OStorageInfo getStorageInfo() {
    return client;
  }

  @Override
  public void replaceStorage(OStorage iNewStorage) {
    throw new UnsupportedOperationException("unsupported replace of storage for remote database");
  }

  private void checkAndSendTransaction() {
    if (this.currentTx.isActive() && ((OTransactionOptimistic) this.currentTx).isChanged()) {
      if (((OTransactionOptimistic) this.getTransaction()).isAlreadyCleared())
        client.reBeginTransaction(getSession(), (OTransactionOptimistic) this.currentTx);
      else client.beginTransaction(getSession(), (OTransactionOptimistic) this.currentTx);
      ((OTransactionOptimistic) this.currentTx).resetChangesTracking();
      ((OTransactionOptimistic) this.currentTx).setSentToServer(true);
    }
  }

  private void fetchTransacion() {
    client.fetchTransaction(this);
  }

  @Override
  public OResultSet query(String query, Object... args) {
    checkOpenness();
    checkAndSendTransaction();
    ORemoteQueryResult result = client.query(this, query, args);
    if (result.isTransactionUpdated()) fetchTransacion();
    if (result.isReloadMetadata()) reload();
    return result.getResult();
  }

  @Override
  public OResultSet query(String query, Map args) {
    checkOpenness();
    checkAndSendTransaction();
    ORemoteQueryResult result = client.query(this, query, args);
    if (result.isTransactionUpdated()) fetchTransacion();
    if (result.isReloadMetadata()) reload();
    return result.getResult();
  }

  @Override
  public OResultSet indexQuery(String indexName, String query, Object... args) {
    checkOpenness();

    if (getTransaction().isActive()) {
      OTransactionIndexChanges changes = getTransaction().getIndexChanges(indexName);
      Set<String> changedIndexes =
          ((OTransactionOptimisticClient) getTransaction()).getIndexChanged();
      if (changedIndexes.contains(indexName) || changes != null) {
        checkAndSendTransaction();
      }
    }
    ORemoteQueryResult result = client.command(this, query, args);
    if (result.isReloadMetadata()) reload();
    return result.getResult();
  }

  @Override
  public OResultSet command(String query, Object... args) {
    checkOpenness();
    checkAndSendTransaction();
    ORemoteQueryResult result = client.command(this, query, args);
    if (result.isTransactionUpdated()) fetchTransacion();
    if (result.isReloadMetadata()) reload();
    return result.getResult();
  }

  @Override
  public OResultSet command(String query, Map args) {
    checkOpenness();
    checkAndSendTransaction();
    ORemoteQueryResult result = client.command(this, query, args);
    if (result.isTransactionUpdated()) fetchTransacion();
    if (result.isReloadMetadata()) reload();
    return result.getResult();
  }

  @Override
  public OResultSet execute(String language, String script, Object... args)
      throws OCommandExecutionException, OCommandScriptException {
    checkOpenness();
    checkAndSendTransaction();
    ORemoteQueryResult result = client.execute(this, language, script, args);
    if (result.isTransactionUpdated()) fetchTransacion();
    if (result.isReloadMetadata()) reload();
    return result.getResult();
  }

  @Override
  public OResultSet execute(String language, String script, Map<String, ?> args)
      throws OCommandExecutionException, OCommandScriptException {
    checkOpenness();
    checkAndSendTransaction();
    ORemoteQueryResult result = client.execute(this, language, script, args);
    if (result.isTransactionUpdated()) fetchTransacion();
    if (result.isReloadMetadata()) reload();
    return result.getResult();
  }

  public void closeQuery(String queryId) {
    client.closeQuery(this, queryId);
    queryClosed(queryId);
  }

  public void fetchNextPage(ORemoteResultSet rs) {
    checkOpenness();
    checkAndSendTransaction();
    client.fetchNextPage(this, rs);
  }

  @Override
  public OLiveQueryMonitor live(String query, OLiveQueryResultListener listener, Object... args) {
    return client.liveQuery(this, query, new OLiveQueryClientListener(this.copy(), listener), args);
  }

  @Override
  public OLiveQueryMonitor live(
      String query, OLiveQueryResultListener listener, Map<String, ?> args) {
    return client.liveQuery(this, query, new OLiveQueryClientListener(this.copy(), listener), args);
  }

  @Override
  public int addBlobCluster(final String iClusterName, final Object... iParameters) {
    int id;
    try (OResultSet resultSet = command("create blob cluster :1", iClusterName)) {
      assert resultSet.hasNext();
      OResult result = resultSet.next();
      assert result.getProperty("value") != null;
      id = result.getProperty("value");
      return id;
    }
  }

  @Override
  public void executeDeleteRecord(OIdentifiable record, int iVersion, boolean iRequired) {
    OTransactionOptimisticClient tx =
        new OTransactionOptimisticClient(this) {
          @Override
          protected void checkTransactionValid() {}
        };
    tx.begin();
    Set<ORecord> records = ORecordInternal.getDirtyManager((ORecord) record).getUpdateRecords();
    if (records != null) {
      for (ORecord rec : records) {
        tx.saveRecord(rec, null, false);
      }
    }
    Set<ORecord> newRecords = ORecordInternal.getDirtyManager((ORecord) record).getNewRecords();
    if (newRecords != null) {
      for (ORecord rec : newRecords) {
        tx.saveRecord(rec, null, false);
      }
    }
    tx.deleteRecord((ORecord) record);
    tx.commit();
  }

  @Override
  public OIdentifiable beforeCreateOperations(OIdentifiable id, String iClusterName) {
    checkSecurity(ORole.PERMISSION_CREATE, id, iClusterName);
    ORecordHook.RESULT res = callbackHooks(ORecordHook.TYPE.BEFORE_CREATE, id);
    if (res == ORecordHook.RESULT.RECORD_CHANGED) {
      if (id instanceof ODocument) {
        ((ODocument) id).validate();
      }
      return id;
    } else if (res == ORecordHook.RESULT.RECORD_REPLACED) {
      ORecord replaced = OHookReplacedRecordThreadLocal.instance().get();
      if (replaced instanceof ODocument) {
        ((ODocument) replaced).validate();
      }
      return replaced;
    }
    return null;
  }

  @Override
  public OIdentifiable beforeUpdateOperations(OIdentifiable id, String iClusterName) {
    checkSecurity(ORole.PERMISSION_UPDATE, id, iClusterName);
    ORecordHook.RESULT res = callbackHooks(ORecordHook.TYPE.BEFORE_UPDATE, id);
    if (res == ORecordHook.RESULT.RECORD_CHANGED) {
      if (id instanceof ODocument) {
        ((ODocument) id).validate();
      }
      return id;
    } else if (res == ORecordHook.RESULT.RECORD_REPLACED) {
      ORecord replaced = OHookReplacedRecordThreadLocal.instance().get();
      if (replaced instanceof ODocument) {
        ((ODocument) replaced).validate();
      }
      return replaced;
    }
    return null;
  }

  @Override
  public void beforeDeleteOperations(OIdentifiable id, String iClusterName) {
    checkSecurity(ORole.PERMISSION_DELETE, id, iClusterName);
    callbackHooks(ORecordHook.TYPE.BEFORE_DELETE, id);
  }

  public void afterUpdateOperations(final OIdentifiable id) {
    callbackHooks(ORecordHook.TYPE.AFTER_UPDATE, id);
    if (id instanceof ODocument) {
      ODocument doc = (ODocument) id;
      OImmutableClass clazz = ODocumentInternal.getImmutableSchemaClass(this, doc);
      if (clazz != null && getTransaction().isActive()) {
        OClassIndexManager.processIndexOnUpdate(this, doc);
      }
    }
  }

  public void afterCreateOperations(final OIdentifiable id) {
    callbackHooks(ORecordHook.TYPE.AFTER_CREATE, id);
    if (id instanceof ODocument) {
      ODocument doc = (ODocument) id;
      OImmutableClass clazz = ODocumentInternal.getImmutableSchemaClass(this, doc);
      if (clazz != null && getTransaction().isActive()) {
        OClassIndexManager.processIndexOnCreate(this, doc);
      }
    }
  }

  public void afterDeleteOperations(final OIdentifiable id) {
    callbackHooks(ORecordHook.TYPE.AFTER_DELETE, id);
    if (id instanceof ODocument) {
      ODocument doc = (ODocument) id;
      OImmutableClass clazz = ODocumentInternal.getImmutableSchemaClass(this, doc);
      if (clazz != null && getTransaction().isActive()) {
        OClassIndexManager.processIndexOnDelete(this, doc);
      }
    }
  }

  @Override
  public boolean beforeReadOperations(OIdentifiable identifiable) {
    return callbackHooks(ORecordHook.TYPE.BEFORE_READ, identifiable) == ORecordHook.RESULT.SKIP;
  }

  @Override
  public void afterReadOperations(OIdentifiable identifiable) {
    callbackHooks(ORecordHook.TYPE.AFTER_READ, identifiable);
  }

  @Override
  public ORecord saveAll(ORecord iRecord, String iClusterName, boolean iForceCreate) {
    OTransactionOptimisticClient tx =
        new OTransactionOptimisticClient(this) {
          @Override
          protected void checkTransactionValid() {}
        };
    tx.begin();
    tx.saveRecord(iRecord, iClusterName, iForceCreate);
    tx.commit();

    return iRecord;
  }

  /**
   * This method is internal, it can be subject to signature change or be removed, do not
   * use. @Internal
   */
  public <RET extends ORecord> RET executeReadRecord(
      final ORecordId rid,
      ORecord iRecord,
      final int recordVersion,
      final String fetchPlan,
      final boolean ignoreCache,
      final boolean iUpdateCache,
      RecordReader recordReader) {
    checkOpenness();
    checkIfActive();

    getMetadata().makeThreadLocalSchemaSnapshot();
    try {

      ORecord record;
      if (!ignoreCache) {
        // SEARCH INTO THE CACHE
        record = getLocalCache().findRecord(rid);
      } else {
        record = null;
      }

      if (record != null) {
        if (iRecord != null && iRecord != record) {
          ORecordInternal.fromStream(iRecord, record.toStream(), this);
          ORecordInternal.setVersion(iRecord, record.getVersion());
          record = iRecord;
        }

        OFetchHelper.checkFetchPlanValid(fetchPlan);
        if (beforeReadOperations(record)) return null;

        if (record.getInternalStatus() == ORecordElement.STATUS.NOT_LOADED) {
          record = reload(record, null, true, true);
        }

        afterReadOperations(record);
        if (record instanceof ODocument) ODocumentInternal.checkClass((ODocument) record, this);
        return (RET) record;
      }

      final ORawBuffer recordBuffer;
      if (!rid.isValid()) recordBuffer = null;
      else {
        OFetchHelper.checkFetchPlanValid(fetchPlan);

        int version;
        if (iRecord != null) version = iRecord.getVersion();
        else version = recordVersion;

        recordBuffer = recordReader.readRecord(rid, fetchPlan, ignoreCache, version);
      }

      if (recordBuffer == null) return null;

      if (iRecord == null || ORecordInternal.getRecordType(iRecord) != recordBuffer.recordType)
        // NO SAME RECORD TYPE: CAN'T REUSE OLD ONE BUT CREATE A NEW ONE FOR IT
        iRecord =
            Orient.instance()
                .getRecordFactoryManager()
                .newInstance(recordBuffer.recordType, rid.getClusterId(), this);

      ORecordInternal.setRecordSerializer(iRecord, getSerializer());
      ORecordInternal.fill(iRecord, rid, recordBuffer.version, recordBuffer.buffer, false, this);

      if (iRecord instanceof ODocument) ODocumentInternal.checkClass((ODocument) iRecord, this);

      if (beforeReadOperations(iRecord)) return null;

      ORecordInternal.fromStream(iRecord, recordBuffer.buffer, this);

      afterReadOperations(iRecord);
      if (iUpdateCache) getLocalCache().updateRecord(iRecord);

      return (RET) iRecord;
    } catch (OOfflineClusterException t) {
      throw t;
    } catch (ORecordNotFoundException t) {
      throw t;
    } catch (Exception t) {
      if (rid.isTemporary())
        throw OException.wrapException(
            new ODatabaseException("Error on retrieving record using temporary RID: " + rid), t);
      else
        throw OException.wrapException(
            new ODatabaseException(
                "Error on retrieving record "
                    + rid
                    + " (cluster: "
                    + getRemoteClient().getPhysicalClusterNameById(rid.getClusterId())
                    + ")"),
            t);
    } finally {
      getMetadata().clearThreadLocalSchemaSnapshot();
    }
  }

  public String getClusterName(final ORecord record) {
    // DON'T ASSIGN CLUSTER WITH REMOTE: SERVER KNOWS THE RIGHT CLUSTER BASED ON LOCALITY
    return null;
  }

  @Override
  public void internalLockRecord(OIdentifiable iRecord, OStorage.LOCKING_STRATEGY lockingStrategy) {
    checkAndSendTransaction();
    ORemoteClient remote = getRemoteClient();
    // -1 value means default timeout
    remote.lockRecord(getSession(), iRecord, lockingStrategy, -1);
  }

  @Override
  public void internalUnlockRecord(OIdentifiable iRecord) {
    ORemoteClient remote = getRemoteClient();
    remote.unlockRecord(getSession(), iRecord.getIdentity());
  }

  @Override
  public <RET extends ORecord> RET lock(ORID recordId) throws OLockException {
    checkOpenness();
    checkIfActive();
    pessimisticLockChecks(recordId);
    checkAndSendTransaction();
    ORemoteClient remote = getRemoteClient();
    // -1 value means default timeout
    OLockRecordResponse response = remote.lockRecord(getSession(), recordId, EXCLUSIVE_LOCK, -1);
    ORecord record =
        fillRecordFromNetwork(
            recordId, response.getRecordType(), response.getVersion(), response.getRecord());
    return (RET) record;
  }

  @Override
  public <RET extends ORecord> RET lock(ORID recordId, long timeout, TimeUnit timeoutUnit)
      throws OLockException {
    checkOpenness();
    checkIfActive();
    pessimisticLockChecks(recordId);
    checkAndSendTransaction();
    ORemoteClient remote = getRemoteClient();
    OLockRecordResponse response =
        remote.lockRecord(getSession(), recordId, EXCLUSIVE_LOCK, timeoutUnit.toMillis(timeout));
    ORecord record =
        fillRecordFromNetwork(
            recordId, response.getRecordType(), response.getVersion(), response.getRecord());
    return (RET) record;
  }

  private ORecord fillRecordFromNetwork(
      ORID recordId, byte recordType, int version, byte[] buffer) {
    beforeReadOperations(recordId);
    ORecord toFillRecord = getLocalCache().findRecord(recordId);
    if (toFillRecord == null)
      toFillRecord =
          Orient.instance()
              .getRecordFactoryManager()
              .newInstance(recordType, recordId.getClusterId(), this);
    ORecordInternal.fill(toFillRecord, recordId, version, buffer, false);
    getLocalCache().updateRecord(toFillRecord);
    afterReadOperations(recordId);
    return toFillRecord;
  }

  @Override
  public void unlock(ORID recordId) throws OLockException {
    checkOpenness();
    checkIfActive();
    internalUnlockRecord(recordId);
  }

  @Override
  public <T> T sendSequenceAction(OSequenceAction action)
      throws ExecutionException, InterruptedException {
    throw new UnsupportedOperationException(
        "Not supported yet."); // To change body of generated methods, choose Tools | Templates.
  }

  public ODatabaseDocumentAbstract delete(final ORecord record) {
    checkOpenness();
    if (record == null) throw new ODatabaseException("Cannot delete null document");
    if (record instanceof OVertex) {
      reload(record, "in*:2 out*:2");
      OVertexDocument.deleteLinks(this, (OVertex) record);
    } else if (record instanceof OEdge) {
      reload(record, "in:1 out:1");
      OEdgeDocument.deleteLinks((OEdge) record);
    }

    try {
      currentTx.deleteRecord(record);
    } catch (OException e) {
      throw e;
    } catch (Exception e) {
      if (record instanceof ODocument)
        throw OException.wrapException(
            new ODatabaseException(
                "Error on deleting record "
                    + record.getIdentity()
                    + " of class '"
                    + ((ODocument) record).getClassName()
                    + "'"),
            e);
      else
        throw OException.wrapException(
            new ODatabaseException("Error on deleting record " + record.getIdentity()), e);
    }
    return this;
  }

  @Override
  public int addCluster(final String iClusterName, final Object... iParameters) {
    checkIfActive();
    return getRemoteClient().addCluster(getSession(), iClusterName, iParameters);
  }

  @Override
  public int addCluster(final String iClusterName, final int iRequestedId) {
    checkIfActive();
    return getRemoteClient().addCluster(getSession(), iClusterName, iRequestedId);
  }

  public ORecordConflictStrategy getConflictStrategy() {
    checkIfActive();
    return getStorageInfo().getRecordConflictStrategy();
  }

  public ODatabaseDocumentAbstract setConflictStrategy(final String iStrategyName) {
    checkIfActive();
    getRemoteClient()
        .setConflictStrategy(
            Orient.instance().getRecordConflictStrategy().getStrategy(iStrategyName));
    return this;
  }

  public ODatabaseDocumentAbstract setConflictStrategy(final ORecordConflictStrategy iResolver) {
    checkIfActive();
    getRemoteClient().setConflictStrategy(iResolver);
    return this;
  }

  /** {@inheritDoc} */
  @Override
  public long countClusterElements(int iClusterId) {
    checkIfActive();
    return getRemoteClient().count(getSession(), iClusterId);
  }

  /** {@inheritDoc} */
  @Override
  public long countClusterElements(int[] iClusterIds) {
    checkIfActive();
    return getRemoteClient().count(getSession(), iClusterIds);
  }

  /** {@inheritDoc} */
  @Override
  public long countClusterElements(final String iClusterName) {
    checkIfActive();

    final int clusterId = getClusterIdByName(iClusterName);
    if (clusterId < 0)
      throw new IllegalArgumentException("Cluster '" + iClusterName + "' was not found");
    return getRemoteClient().count(getSession(), clusterId);
  }

  @Override
  public long getClusterRecordSizeByName(final String clusterName) {
    checkIfActive();
    try {
      return getRemoteClient().getClusterRecordsSizeByName(clusterName);
    } catch (Exception e) {
      throw OException.wrapException(
          new ODatabaseException("Error on reading records size for cluster '" + clusterName + "'"),
          e);
    }
  }

  @Override
  public boolean dropCluster(final String iClusterName) {
    checkIfActive();
    final int clusterId = getClusterIdByName(iClusterName);
    OSessionSchema schema = metadata.getSchema();
    OClass clazz = schema.getClassByClusterId(clusterId);
    if (clazz != null) clazz.removeClusterId(clusterId);
    if (schema.getBlobClusters().contains(clusterId)) schema.removeBlobCluster(iClusterName);
    getLocalCache().freeCluster(clusterId);
    checkForClusterPermissions(iClusterName);
    return getRemoteClient().dropCluster(getSession(), iClusterName);
  }

  @Override
  public boolean dropCluster(final int clusterId) {
    checkIfActive();

    OSessionSchema schema = metadata.getSchema();
    final OClass clazz = schema.getClassByClusterId(clusterId);
    if (clazz != null) clazz.removeClusterId(clusterId);
    getLocalCache().freeCluster(clusterId);
    if (schema.getBlobClusters().contains(clusterId))
      schema.removeBlobCluster(getClusterNameById(clusterId));

    checkForClusterPermissions(getClusterNameById(clusterId));

    final String clusterName = getClusterNameById(clusterId);
    if (clusterName == null) {
      return false;
    }

    final ORecordIteratorCluster<ODocument> iteratorCluster = browseCluster(clusterName);
    if (iteratorCluster == null) {
      return false;
    }

    while (iteratorCluster.hasNext()) {
      final ODocument document = iteratorCluster.next();
      this.delete(document);
    }

    return getRemoteClient().dropCluster(getSession(), clusterId);
  }

  public boolean dropClusterInternal(int clusterId) {
    return getRemoteClient().dropCluster(getSession(), clusterId);
  }

  @Override
  public long getClusterRecordSizeById(final int clusterId) {
    checkIfActive();
    try {
      return getRemoteClient().getClusterRecordsSizeById(clusterId);
    } catch (Exception e) {
      throw OException.wrapException(
          new ODatabaseException(
              "Error on reading records size for cluster with id '" + clusterId + "'"),
          e);
    }
  }

  @Override
  public long getSize() {
    checkIfActive();
    return getRemoteClient().getSize(getSession());
  }

  @Override
  public <DB extends ODatabaseDocument> DB checkSecurity(
      ORule.ResourceGeneric resourceGeneric, String resourceSpecific, int iOperation) {
    return (DB) this;
  }

  @Override
  public <DB extends ODatabaseDocument> DB checkSecurity(
      ORule.ResourceGeneric iResourceGeneric, int iOperation, Object iResourceSpecific) {
    return (DB) this;
  }

  @Override
  public <DB extends ODatabaseDocument> DB checkSecurity(
      ORule.ResourceGeneric iResourceGeneric, int iOperation, Object... iResourcesSpecific) {
    return (DB) this;
  }

  @Override
  public boolean isRemote() {
    return true;
  }

  @Override
  public String incrementalBackup(final String path) throws UnsupportedOperationException {
    checkOpenness();
    checkIfActive();
    checkSecurity(ORule.ResourceGeneric.DATABASE, "backup", ORole.PERMISSION_EXECUTE);

    return getRemoteClient().incrementalBackup(getSession(), path, null);
  }

  @Override
  public ORecordMetadata getRecordMetadata(final ORID rid) {
    checkIfActive();
    return getRemoteClient().getRecordMetadata(getSession(), rid);
  }

  /** {@inheritDoc} */
  @Override
  public void freeze(final boolean throwException) {
    checkOpenness();
    logger.error(
        "Only local paginated storage supports freeze. If you are using remote client please"
            + " use OrientDB instance instead",
        null);

    return;
  }

  /** {@inheritDoc} */
  @Override
  public void freeze() {
    freeze(false);
  }

  /** {@inheritDoc} */
  @Override
  public void release() {
    checkOpenness();
    logger.error(
        "Only local paginated storage supports release. If you are using remote client please"
            + " use OrientDB instance instead",
        null);
    return;
  }

  @Override
  public List<String> backup(
      final OutputStream out,
      final Map<String, Object> options,
      final Callable<Object> callable,
      final OCommandOutputListener iListener,
      final int compressionLevel,
      final int bufferSize)
      throws IOException {
    throw new UnsupportedOperationException(
        "backup is not supported against remote storage. Use OrientDB instance command instead");
  }

  @Override
  public void restore(
      final InputStream in,
      final Map<String, Object> options,
      final Callable<Object> callable,
      final OCommandOutputListener iListener)
      throws IOException {
    throw new UnsupportedOperationException(
        "restore is not supported against remote instance. Use OrientDB instance command instead");
  }

  /** {@inheritDoc} */
  public OSBTreeCollectionManager getSbTreeCollectionManager() {
    return getRemoteClient().getSBtreeCollectionManager();
  }

  @Override
  public void reload() {
    checkIfActive();

    if (this.isClosed()) {
      throw new ODatabaseException("Cannot reload a closed db");
    }
    metadata.reload();
    getRemoteClient().reload(getSession());
  }

  @Override
  public void internalCommit(OTransactionInternal transaction) {
    this.getRemoteClient().commit(this, transaction);
  }

  @Override
  public boolean isClosed() {
    return status == STATUS.CLOSED || getRemoteClient().isClosed(getSession());
  }

  public void internalClose(boolean recycle) {
    if (status != STATUS.OPEN) return;

    checkIfActive();

    try {
      closeActiveQueries();
      localCache.shutdown();

      if (isClosed()) {
        status = STATUS.CLOSED;
        sharedContext.endSession();
        return;
      }

      try {
        rollback(true);
      } catch (Exception e) {
        logger.error("Exception during rollback of active transaction", e);
      }

      callOnCloseListeners();

      status = STATUS.CLOSED;
      sharedContext.endSession();
      if (!recycle) {
        sharedContext = null;

        if (getRemoteClient() != null) getRemoteClient().close(getSession());
      }

    } finally {
      // ALWAYS RESET TL
      ODatabaseRecordThreadLocal.instance().remove();
    }
  }

  @Override
  public long[] getClusterDataRange(int currentClusterId) {
    return getRemoteClient().getClusterDataRange(getSession(), currentClusterId);
  }

  @Override
  public void setDefaultClusterId(int addCluster) {
    getRemoteClient().setDefaultClusterId(addCluster);
  }

  @Override
  public long getLastClusterPosition(int clusterId) {
    throw new UnsupportedOperationException();
  }

  @Override
  public String getClusterRecordConflictStrategy(int clusterId) {
    throw new UnsupportedOperationException();
  }

  public OTransactionOptimisticClient getActiveTx() {
    if (currentTx.isActive()) {
      return (OTransactionOptimisticClient) currentTx;
    } else {
      currentTx = new OTransactionOptimisticClient(this);
      currentTx.begin();
      return (OTransactionOptimisticClient) currentTx;
    }
  }

  @Override
  public int[] getClustersIds(Set<String> filterClusters) {
    checkIfActive();
    return filterClusters.stream().map((c) -> getClusterIdByName(c)).mapToInt(i -> i).toArray();
  }

  /** {@inheritDoc} */
  @Override
  public void truncateCluster(String clusterName) {
    command("truncate cluster " + clusterName).close();
  }

  @Override
  public void truncateClass(String name) {
    command("truncate class " + name).close();
  }

  @Override
  public long truncateClusterInternal(String name) {
    throw new UnsupportedOperationException();
  }

  @Override
  public long truncateClass(String name, boolean polimorfic) {
    long count = 0;
    if (polimorfic) {
      try (OResultSet result = command("truncate class " + name + " polymorphic ")) {
        while (result.hasNext()) {
          count += (long) result.next().getProperty("count");
        }
      }
    } else {
      try (OResultSet result = command("truncate class " + name)) {
        while (result.hasNext()) {
          count += (long) result.next().getProperty("count");
        }
      }
    }
    return count;
  }

  @Override
  public void commitPreallocate() {
    throw new UnsupportedOperationException();
  }

  @Override
  public void internalCommitPreallocate(OTransactionOptimistic oTransactionOptimistic) {
    throw new UnsupportedOperationException();
  }

  @Override
  public void remoteRollback(OTransactionOptimistic oTransactionOptimistic) {
    client.rollback(getSession(), oTransactionOptimistic);
  }

  public OPhysicalPosition[] higherPhysicalPositions(
      int clusterId, OPhysicalPosition physicalPosition) {
    return getRemoteClient().higherPhysicalPositions(getSession(), clusterId, physicalPosition);
  }

  public OPhysicalPosition[] lowerPhysicalPositions(
      int clusterId, OPhysicalPosition physicalPosition) {
    return getRemoteClient().lowerPhysicalPositions(getSession(), clusterId, physicalPosition);
  }

  public OPhysicalPosition[] ceilingPhysicalPositions(
      int clusterId, OPhysicalPosition physicalPosition) {
    return getRemoteClient().ceilingPhysicalPositions(getSession(), clusterId, physicalPosition);
  }

  public OPhysicalPosition[] floorPhysicalPositions(
      int clusterId, OPhysicalPosition physicalPosition) {
    return getRemoteClient().floorPhysicalPositions(getSession(), clusterId, physicalPosition);
  }

  @Override
  public long countRecords() {
    return getRemoteClient().countRecords(getSession());
  }

  @Override
  public boolean isReusable() {
    return !getRemoteClient().isClosed(getSession());
  }

  @Override
  public OBonsaiCollectionPointer createSBTree(int clusterId, UUID ownerUUID) {
    throw new UnsupportedOperationException();
  }

  public ORawBuffer directRead(
      ORecordId rid, String fetchPlan, boolean ignoreCache, int recordVersion) {
    return getRemoteClient().readRecord(getSession(), rid, fetchPlan, ignoreCache, ignoreCache);
  }

  public ORawBuffer readIfVersionIsNotLatest(
      ORecordId rid, String fetchPlan, boolean ignoreCache, int recordVersion) {
    return getRemoteClient()
        .readRecordIfVersionIsNotLatest(getSession(), rid, fetchPlan, ignoreCache, recordVersion);
  }
}
